#include <atomic>
#include <chrono>
#include <climits>
#include <condition_variable> 
#include <functional> 
#include <iomanip> 
#include <iostream> 
#include <memory> 
#include <mutex> 
#include <optional> 
#include <queue> 
#include <sstream> 
#include <string> 
#include <thread> 
#include <utility> 
#include <vector> 

namespace my_poll {
    struct base_msg {
        virtual ~base_msg() = default;
    };
    
    template<typename Msg>
    struct pack_msg : public base_msg {
        Msg pack;
        explicit pack_msg(const Msg& msg) : pack(msg) {} 
    };
    
    class queue {
        std::mutex mtx; // 修正：移除多余空格
        std::condition_variable var;
        std::queue<std::shared_ptr<base_msg>> msg_queue;
        
    public:
        template<typename T>
        void push(const T& msg) {
            std::lock_guard<std::mutex> lock(mtx);
            msg_queue.push(std::make_shared<pack_msg<T>>(msg));
            var.notify_one();
        }
        
        std::shared_ptr<base_msg> wait_pop() {
            std::unique_lock<std::mutex> lock(mtx);
            var.wait(lock, [this] { return !msg_queue.empty(); });
            auto msg = msg_queue.front();
            msg_queue.pop();
            return msg;
        }

        bool try_pop(std::shared_ptr<base_msg>& msg) {
            std::lock_guard<std::mutex> lock(mtx);
            if (msg_queue.empty()) return false;
            msg = msg_queue.front();
            msg_queue.pop();
            return true;
        }

        bool empty() {
            std::lock_guard<std::mutex> lock(mtx);
            return msg_queue.empty();
        }
        
        size_t size() {
            std::lock_guard<std::mutex> lock(mtx);
            return msg_queue.size();
        }
    };

    class sender {
        queue *task;
    public:
        sender() : task(nullptr) {}
        sender(queue *q) : task(q) {}
        
        template<typename Msg>
        void send(const Msg& msg) const {
            if (task) task->push(msg);
        }
    };
    
    struct close_queue {};

    template<typename PreviousDispatcher, typename Msg, typename Func>
    class TemplateDispatcher {
        queue *task;
        PreviousDispatcher *prev;
        Func f; 
        bool is_link;

        TemplateDispatcher(const TemplateDispatcher &) = delete; 
        TemplateDispatcher &operator=(const TemplateDispatcher &) = delete; 
        
        template<typename Dispatcher, typename OtherMsg, typename OtherFunc>
        friend class TemplateDispatcher; // 允许同模板不同参数的类访问

        bool dispatch(const std::shared_ptr<base_msg> &msg) {
            if (auto wrapped = dynamic_cast<pack_msg<Msg> *>(msg.get())) {
                f(wrapped->pack); // 修正：直接调用而不是解引用
                return true;
            } else {
                return prev->dispatch(msg);
            }
        }

        void wait_dispatch() {
            for (;;) {
                auto msg = task->wait_pop();
                if (dispatch(msg)) break;
            }
        }

    public:
        TemplateDispatcher(TemplateDispatcher &&other)
        : task(other.task), prev(other.prev), f(std::move(other.f)), is_link(other.is_link) {
            other.is_link = true;
        }
        
        
        TemplateDispatcher(queue *q_, PreviousDispatcher *prev_, Func &&f_)
        : task(q_), prev(prev_), f(std::forward<Func>(f_)), is_link(false) {
            prev_->is_link = true; 
        }
        
        template <typename OtherMsg, typename OtherFunc>
        TemplateDispatcher<TemplateDispatcher, OtherMsg, OtherFunc> handle(OtherFunc &&of) {
        
            return TemplateDispatcher<TemplateDispatcher, OtherMsg, OtherFunc>(this->task, this, std::forward<OtherFunc>(of));
        }

        ~TemplateDispatcher() noexcept(false) {
            if (!is_link) {
                wait_dispatch();
            }
        }
    };

    class dispatcher {
        queue *q;
        bool is_link;

        dispatcher(const dispatcher &) = delete;
        dispatcher &operator=(const dispatcher &) = delete;

        template <typename Dispatcher, typename Msg, typename Func>
        friend class TemplateDispatcher;

        void wait_and_dispatch() {
            for (;;) {
                auto msg = q->wait_pop();
                dispatch(msg);
            }
        }

        bool dispatch(const std::shared_ptr<base_msg> &msg) {
            if (dynamic_cast<pack_msg<close_queue> *>(msg.get())) throw close_queue();
            return false;
        }

    public:
        dispatcher(dispatcher &&other) : q(other.q), is_link(other.is_link) { 
            other.is_link = true; 
        }
        
        explicit dispatcher(queue *q_) : q(q_), is_link(false) {}

        template <typename Message, typename Func>
        TemplateDispatcher<dispatcher, Message, Func> handle(Func &&f) {
            return TemplateDispatcher<dispatcher, Message, Func>(q, this, std::forward<Func>(f));
        }

        ~dispatcher() noexcept(false) {
            if (!is_link) wait_and_dispatch();
        }
    };

    class receiver {
        queue q;
    public:
        operator sender() { return sender(&q); }
        dispatcher wait() { return dispatcher(&q); }
    };
}

namespace act_pool {
    using Clock = std::chrono::steady_clock;
    using Ms = std::chrono::milliseconds;
    
    // 取消令牌
    struct cancell_token {
        std::atomic<bool> cancelled{false};
        void cancel() { cancelled.store(true, std::memory_order_relaxed); }
        bool is_cancelled() const { return cancelled.load(std::memory_order_acquire); }
    };
    
    // 提交任务消息
    struct submit_task {
        std::string id; //任务标识
        int priority; // 优先级，越大越优先执行
        Ms queue_timeout; // 队列允许等待最长时间
        Ms exec_timeout; // 执行允许最长时间
        std::function<void(cancell_token&)> task; // 任务函数
        my_poll::sender reply_to; // 回复发送者
    };
    
    // 任务开始事件
    struct start_task {
        std::string id;
    };

    // 任务完成事件
    struct finish_task {
        std::string id;
        Ms run_time;
    };

    // 排队超时事件
    struct queue_timeout_task {
        std::string id;
    };

    // 执行超时事件
    struct exec_timeout_task {
        std::string id;
    };

    // 停止线程池事件
    struct stop_pool {};

    struct ScheduledTask {
        std::string id;
        int priority;
        Clock::time_point enqueued_at;
        Ms queue_timeout;
        Ms exec_timeout;
        std::function<void(cancell_token&)> func; //任务
        my_poll::sender reply_to; //结果回收目标
        std::shared_ptr<cancell_token> token; //取消令牌
    };

    struct TaskCompare {
        bool operator()(const std::shared_ptr<ScheduledTask> &a, const std::shared_ptr<ScheduledTask> &b) const {
            if (a->priority == b->priority) return a->enqueued_at > b->enqueued_at;
            return a->priority < b->priority;
        }
    };

    class thread_pool {
        my_poll::receiver incoming;
        std::vector<std::thread> workers;
        std::thread control_thread;
        
        // 任务优先队列
        std::priority_queue<std::shared_ptr<ScheduledTask>, 
        std::vector<std::shared_ptr<ScheduledTask>>, TaskCompare> pq;
        std::mutex pq_mutex;
        std::condition_variable pq_cv;
        bool stopping{false};

        void control_loop() {
            try {
                for (;;) {
                    incoming.wait()
                        .handle<submit_task>([&](const submit_task &msg) {
                            auto t = std::make_shared<ScheduledTask>();
                            t->id = msg.id;
                            t->priority = msg.priority;
                            t->enqueued_at = Clock::now();
                            t->queue_timeout = msg.queue_timeout;
                            t->exec_timeout = msg.exec_timeout;
                            t->func = msg.task;
                            t->reply_to = msg.reply_to;
                            t->token = std::make_shared<cancell_token>();
                            {
                                std::lock_guard<std::mutex> lk(pq_mutex);
                                pq.push(t);
                            }
                            pq_cv.notify_one();
                        })
                        .handle<stop_pool>([&](const stop_pool &) {
                            {
                                std::lock_guard<std::mutex> lk(pq_mutex);
                                stopping = true;
                            }
                            pq_cv.notify_all();
                            throw my_poll::close_queue();
                        })
                        .handle<my_poll::close_queue>([&](const my_poll::close_queue&) {
                            throw my_poll::close_queue();
                        });
                }
            } catch (const my_poll::close_queue&) {
                // 正常退出
            }
        }

        // 工作线程主循环
        void work_loop(size_t worker_index) {
            (void)worker_index; // 消除未使用参数警告
            
            for (;;) {
                std::shared_ptr<ScheduledTask> task;
                {
                    std::unique_lock<std::mutex> lk(pq_mutex);
                    pq_cv.wait(lk, [&] { return stopping || !pq.empty(); });
                    if (stopping && pq.empty()) return;
                    task = pq.top();
                    pq.pop();
                }

                // 队列超时检查
                if (task->queue_timeout.count() > 0) {
                    const auto waited = std::chrono::duration_cast<Ms>(Clock::now() - task->enqueued_at);
                    if (waited > task->queue_timeout) {
                        // 将任务优先级提到最高并重新入队，让其它线程优先处理
                        task->priority = INT_MAX;
                        task->enqueued_at = Clock::now(); // 重要：更新入队时间，避免无限循环
                        {
                            std::lock_guard<std::mutex> lk(pq_mutex);
                            pq.push(task);
                        }
                        pq_cv.notify_one();
                        continue;
                    }
                }
                
                // 执行超时看门狗
                std::atomic<bool> finished{false};
                std::atomic<bool> reported{false};
                std::mutex watch_mutex;
                std::condition_variable watch_cv;

                std::thread watchdog;
                if (task->exec_timeout.count() > 0) {
                    watchdog = std::thread([&, token = task->token, id = task->id, exec_to = task->exec_timeout] {
                        std::unique_lock<std::mutex> lk(watch_mutex);
                        if (watch_cv.wait_for(lk, exec_to, [&] { return finished.load(std::memory_order_acquire); })) {
                            return;
                        }
                        token->cancel();
                        bool expected = false;
                        if (reported.compare_exchange_strong(expected, true)) {
                            task->reply_to.send(exec_timeout_task{id});
                        }
                    });
                }
                
                // 上报任务开始
                task->reply_to.send(start_task{task->id});
                auto start = Clock::now();
                
                // 执行任务
                try {
                    task->func(*task->token);
                } catch (...) {
                    // 忽略任务异常
                }
                
                const auto run_time = std::chrono::duration_cast<Ms>(Clock::now() - start);

                // 标记任务完成并唤醒看门狗
                {
                    finished.store(true, std::memory_order_release);
                    watch_cv.notify_all();
                }
                if (watchdog.joinable()) watchdog.join();

                bool expected = false;
                if (!task->token->is_cancelled() && reported.compare_exchange_strong(expected, true)) {
                    task->reply_to.send(finish_task{task->id, run_time});
                }
            }
        }

    public:
        explicit thread_pool(size_t thread_count) {
            workers.reserve(thread_count);
            control_thread = std::thread([this] { control_loop(); });
            for (size_t i = 0; i < thread_count; ++i) {
                workers.emplace_back([this, i] { work_loop(i); });
            }
        }

        ~thread_pool() {
            stop();
        }
        
        void stop() {
            {
                std::lock_guard<std::mutex> lk(pq_mutex);
                if (stopping && !control_thread.joinable() && workers.empty()) return;
            }
            my_poll::sender in = incoming;
            in.send(stop_pool{});
            in.send(my_poll::close_queue{});

            if (control_thread.joinable()) control_thread.join();
            {
                std::lock_guard<std::mutex> lk(pq_mutex);
                stopping = true;
            }
            pq_cv.notify_all();
            for (auto &t : workers) {
                if (t.joinable()) t.join();
            }
            workers.clear();
        }

        my_poll::sender get_sender() { return incoming; }
    };
}

// 结果记录演员：打印线程池事件
class result_logger {
    my_poll::receiver incoming;
    std::thread t;

public:
    result_logger() {
        t = std::thread([this] { run(); });
    }

    ~result_logger() {
        stop();
    }

    my_poll::sender get_sender() { return incoming; }

    void stop() {
        if (t.joinable()) {
            my_poll::sender in = incoming;
            in.send(my_poll::close_queue{});
            t.join();
        }
    }

private:
    static std::string now_ts() {
        using Clock = std::chrono::system_clock;
        auto now = Clock::to_time_t(Clock::now());
        std::tm tm{};
        #ifdef _WIN32
        localtime_s(&tm, &now);
        #else
        localtime_r(&now, &tm);
        #endif
        std::ostringstream oss;
        oss << std::put_time(&tm, "%H:%M:%S");
        return oss.str();
    }

    void run() {
        try {
            for (;;) {
                incoming
                    .wait()
                    .handle<act_pool::start_task>([&](const act_pool::start_task &m) {
                        std::cout << "[" << now_ts() << "] 开始: " << m.id << std::endl;
                    })
                    .handle<act_pool::finish_task>([&](const act_pool::finish_task &m) {
                        std::cout << "[" << now_ts() << "] 完成: " << m.id
                                  << "，耗时毫秒=" << m.run_time.count() << std::endl;
                    })
                    .handle<act_pool::exec_timeout_task>([&](const act_pool::exec_timeout_task &m) {
                        std::cout << "[" << now_ts() << "] 执行超时: " << m.id << std::endl;
                    })
                    .handle<act_pool::queue_timeout_task>([&](const act_pool::queue_timeout_task &m) {
                        std::cout << "[" << now_ts() << "] 排队超时: " << m.id << std::endl;
                    })
                    .handle<my_poll::close_queue>([](const my_poll::close_queue &) {
                        throw my_poll::close_queue();
                    });
            }
        } catch (const my_poll::close_queue &) {
            // 正常退出
        }
    }
};

// 协作式 sleep：分片休眠并检查取消令牌
static void cooperative_sleep(act_pool::cancell_token& token, std::chrono::milliseconds total) {
    const auto slice = std::chrono::milliseconds(25);
    auto elapsed = std::chrono::milliseconds(0);
    while (elapsed < total && !token.is_cancelled()) {
        std::this_thread::sleep_for(slice);
        elapsed += slice;
    }
}

